#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>

#include "mqtt_communication.h"
#include "posix_sockets.h"
#include "global_val.h"

static struct mqtt_client mqtt_communication_client;
uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
static int sockfd;
const char* client_id = NULL;
pthread_t mqtt_client_id;

int mqtt_communication_init(const char *addr, const char *port)
{
	int res;
	
	sockfd = open_nb_socket(addr, port);
	if (sockfd == -1) {
        perror("Failed to open socket: ");
        exit_example(EXIT_FAILURE, sockfd, NULL);
    }
	
	//callback void (*publish_response_callback)(void** state,struct mqtt_response_publish *publish) 接收消息的回调函数
	mqtt_init(&mqtt_communication_client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
	
	uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
	mqtt_connect(&mqtt_communication_client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
    if (mqtt_communication_client.error != MQTT_OK) {
        fprintf(stderr, "error: %s\n", mqtt_error_str(mqtt_communication_client.error));
        exit_example(EXIT_FAILURE, sockfd, NULL);
    }
	
    if(pthread_create(&mqtt_client_id, NULL, client_refresher, &mqtt_communication_client)) {
        fprintf(stderr, "Failed to start client pthread\n");
        exit_example(EXIT_FAILURE, sockfd, NULL);
    }
	
	pthread_detach(mqtt_client_id);
	
	return 0;
}

void publish_callback(void** unused, struct mqtt_response_publish *published) 
{
    /* not used in this example */
	
	printf("Received publish('%s'): %s\n", (const char *)published->topic_name, (const char*) published->application_message);
	
}

void subscription_topic(const char *topic)
{
	mqtt_subscribe(&mqtt_communication_client, topic, 0);
}

//消息 长度 主题 发送等级
void public_message(const char *message, const int message_length, const char *topic, uint8_t publish_flags)
{
	mqtt_publish(&mqtt_communication_client, topic, message, message_length, publish_flags);

	/* check for errors */
	if (mqtt_communication_client.error != MQTT_OK) {
		fprintf(stderr, "error: %s\n", mqtt_error_str(mqtt_communication_client.error));
		exit_example(EXIT_FAILURE, sockfd, &mqtt_client_id);
	}
}

void* client_refresher(void* client)
{
    while(1) 
    {
        mqtt_sync((struct mqtt_client*) client);
        usleep(100000U);
    }
    return NULL;
}

void exit_example(int status, int sockfd, pthread_t *client_daemon)
{
    if (sockfd != -1) close(sockfd);
    if (client_daemon != NULL) pthread_cancel(*client_daemon);
    exit(status);
}